package com.wu.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author wufei
 * @date 2018/3/21/021 19:09
 */
public class KafkaConsume {

    //High lever consumer
    //Low lever consumer

    private final static String BORKER_LIST = "47.98.227.22:9092,47.98.186.174:9092";

    KafkaConsumer<Integer,String> kafkaConsumer;

    public KafkaConsume() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,BORKER_LIST);

        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"Group1");
        //是否自动提交 offset
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
        //自动提交的间隔时间
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
        //设置使用最开始的offset偏移量为当前groupid的最早消息
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        //心跳时间
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"30000");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        kafkaConsumer = new KafkaConsumer<Integer, String>(properties);
    }

    public KafkaConsumer<Integer, String> getKafkaConsumer() {
        return kafkaConsumer;
    }

    public static void main(String[] args) {
        System.out.printf(Math.abs("Group1".hashCode()%50)+"");
    }
}
